package com.atguigu.api2

import com.atguigu.api.SensorReading
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector


/**
 * @description: 创建不同类型的窗口,窗口必须跟着后面函数才算是完整的一次计算
 * @time: 2020/6/21 13:10
 * @author: baojinlong
 **/
object WindowApiTest {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)
    // 设置处理时间的维度 加上后面给分配时间,同时加.assignAscendingTimestamps(_.timestamp * 1000L)提取时间字段
    // environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    // environment.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime) // 默认是事件处理时间
    val inputStreamFromFile: DataStream[String] = environment.readTextFile("E:/qj_codes/big-data/FlinkTutorial/src/main/resources/sensor.data")

    // 基本转换操作
    val dataStream: DataStream[SensorReading] = inputStreamFromFile
      .map(data => {
        val dataArray: Array[String] = data.split(",")
        SensorReading(dataArray(0), dataArray(1).toLong, dataArray(2).toDouble)
      })
    // .assignAscendingTimestamps(_.timestamp * 1000L) // 提取升序的时间戳

    val resultDataStream: DataStream[(Long, Int)] = dataStream
      .keyBy("id")
      //.window(EventTimeSessionWindows.withGap(Time.minutes(1))) // 会话窗口
      //.timeWindow(Time.hours(1), Time.minutes(1)) // 传一个参数为滚动窗口,两个为滑动窗口
      //.window(TumblingEventTimeWindows.of(Time.days(1), Time.hours(2))) // 滚动窗口
      //.countWindow(10, 2) // 计数窗口
      //.timeWindow(Time.milliseconds(1000))
      // 注意如果时间没达到则没有数据输出
      //.timeWindow(Time.seconds(15))
      .timeWindow(Time.seconds(15), Time.seconds(5)) // 滑动窗口
      // .reduce(new MyReduce) // 增量计算窗口
      .apply(new MyWindowFunction3) //全窗口案例
    // 默认是处理时间
    dataStream
      .keyBy("id")
      //.window(TumblingEventTimeWindows.of(Time.seconds(15))) // 滚动时间窗口
      // .window(SlidingProcessingTimeWindows.of(Time.seconds(15),Time.seconds(3))) // 滑动窗口
      // .window(EventTimeSessionWindows.withGap(Time.seconds(10))) // 滚动计数窗口
      // .countWindow(10) // 滚动计数窗口
      //.countWindow(10, 2) // 滑动计数窗口
      .timeWindow(Time.seconds(15)) // 滚动窗口


    // 打印结果
    dataStream.print("dataStream")
    resultDataStream.print("resultDataStream")
    environment.execute("sink simple test job")
  }
}

// 自定义全窗口函数
class MyWindowFunction3 extends WindowFunction[SensorReading, (Long, Int), Tuple, TimeWindow] {
  override def apply(key: Tuple, window: TimeWindow, input: Iterable[SensorReading], out: Collector[(Long, Int)]): Unit = {
    out.collect((window.getStart, input.size))
  }
}
